package com.wyz.socket.core.impl.dispatcher.bridge;


import com.wyz.socket.core.api.Receiver;
import com.wyz.socket.core.api.Sender;
import com.wyz.socket.core.api.dispatcher.ReceiveDispatcher;
import com.wyz.socket.core.api.dispatcher.SendDispatcher;
import com.wyz.socket.core.base.IoArgs;
import com.wyz.socket.core.base.exceptions.IoArgsException;
import com.wyz.socket.core.base.packet.AbstractSendPacket;
import com.wyz.socket.core.utils.Log;
import com.wyz.socket.core.utils.buffer.CircularByteBuffer;

import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.concurrent.atomic.AtomicBoolean;

public class BridgeSocketDispatcher implements SendDispatcher, ReceiveDispatcher {

    private final CircularByteBuffer mBuffer = new CircularByteBuffer(512, true);

    private final ReadableByteChannel readChannel = Channels.newChannel(mBuffer.getInputStream());
    private final WritableByteChannel writeChannel = Channels.newChannel(mBuffer.getOutputStream());

    private final IoArgs receiveIoArgs = new IoArgs(256, false);
    private final IoArgs sendIoArgs = new IoArgs();

    private final Receiver receiver;
    private volatile Sender sender;

    private final AtomicBoolean isSending = new AtomicBoolean();

    private final IoArgs.IoBufferEventProcessor senderEventProcessor = new IoArgs.IoBufferEventProcessor() {
        @Override
        public void onstart() {

        }

        @Override
        public IoArgs providerIoArgs() {
            try {
                int available = mBuffer.getAvailable();
                IoArgs sendIoArgs = BridgeSocketDispatcher.this.sendIoArgs;
                if (available > 0) {
                    sendIoArgs.limit(available);
                    sendIoArgs.startWriting();
                    sendIoArgs.readFrom(readChannel);
                    sendIoArgs.endWriting();
                    return sendIoArgs;
                }
            } catch (Exception e) {
                Log.printException("send Processor", "Processor providerIoArgs ", e);
            }
            return null;
        }

        @Override
        public boolean onConsumeFailed(Throwable e) {
            Log.printException("BridgeSocketDispatcher", "senderEventProcessor onConsumeFailed "+e.getClass().getSimpleName(), e);
            if (e instanceof IoArgsException) {
                synchronized (isSending) {
                    isSending.set(false);
                }
                requesterSend();
                return false;
            } else {
                return true;
            }
        }

        @Override
        public boolean onConsumeCompleted(IoArgs args) {
            synchronized (isSending) {
                isSending.set(false);
            }
            requesterSend();
            return false;
        }

        @Override
        public void onEnd() {

        }
    };

    private final IoArgs.IoBufferEventProcessor receiverEventProcessor = new IoArgs.IoBufferEventProcessor() {
        @Override
        public void onstart() {

        }

        @Override
        public IoArgs providerIoArgs() {
            receiveIoArgs.resetLimit();
            receiveIoArgs.startWriting();
            return receiveIoArgs;
        }

        @Override
        public boolean onConsumeFailed(Throwable e) {
            Log.printlnInfo("BridgeSocketDispatcher receiverEventProcessor onConsumeFailed "+e.getClass().getSimpleName() , e.getMessage());
            if (e instanceof IoArgsException) {
                return false;
            } else {
                return true;
            }
        }

        @Override
        public boolean onConsumeCompleted(IoArgs args) {
            args.endWriting();
            try {
                args.writeTo(writeChannel);
                requesterSend();
                return true;
            } catch (Exception e) {
                Log.printException("receive Event", " Processor onConsumeCompleted", e);
                return false;
            }
        }

        @Override
        public void onEnd() {

        }
    };

    public BridgeSocketDispatcher(Receiver receiver) {
        this.receiver = receiver;
        this.receiver.registerSocketListener(this);
        this.receiver.setReceiveListener(receiverEventProcessor);
        Log.println("BridgeSocketDispatcher" ,"BridgeSocketDispatcher create" , " change model to Bridge");
    }

    public void bindSender(Sender sender) {
        final Sender oldSender = this.sender;
        if (oldSender != null) {
            oldSender.setSendListener(null);
            oldSender.unregisterSocketListener(this);
        }

        synchronized (isSending) {
            isSending.set(false);
        }
        mBuffer.clear();

        this.sender = sender;
        if (sender != null) {
            sender.setSendListener(senderEventProcessor);
            sender.registerSocketListener(this);
            requesterSend();
        }
    }

    private void requesterSend() {
        synchronized (isSending) {
            Sender sender = this.sender;
            if (sender == null || isSending.get()) {
                return;
            }
            if (mBuffer.getAvailable() > 0) {
                try {
                    isSending.set(true);
                    sender.postSendAsync();
                } catch (Exception e) {
                    Log.printException("Bridge socket Dispatcher", "requesterSend  create exception ", e);
                    e.printStackTrace();
                    try {
                        this.close();
                    } catch (IOException ioException) {
                        ioException.printStackTrace();
                    }
                }
            }
        }
    }


    private void requesterReceive() {
        try {
            receiver.postReceiveAsync();
        } catch (Exception e) {
            Log.printException("Bridge socket Dispatcher", "requesterReceive  create exception ", e);
        }
    }

    // receiver dispatcher
    @Override
    public void start() {
        requesterReceive();
    }

    @Override
    public void stop() {
        receiver.setReceiveListener(null);
    }

    // send dispatcher
    @Override
    public void send(AbstractSendPacket<?> packet) {
        requesterSend();
    }

    @Override
    public void cancel(AbstractSendPacket<?> packet) {

    }

    // send heartbeat
    @Override
    public void sendHeartbeat() {

    }

    @Override
    public long getLastActiveTime() {
        return 0;
    }


    //close
    @Override
    public void close() throws IOException {

    }

    @Override
    public <T> void onClose(T t) {
        if (t instanceof SocketChannel) {
            Log.printlnInfo("BridgeSocketDispatcher", "SocketChannel is Closed");
        }
    }
}
